from _datetime import datetime
from es_pandas import es_pandas

from air_web.config.config import config
from air_web.data_platform import sql_engine
from air_web.dw.modify_alias import modify_alias


def get_on7_list():
    sql = "select distinct on7 from {c_cons_all}".format(c_cons_all=c_cons_all)
    on7_df = sql_engine.query(sql)
    on7_list = on7_df["on7"].tolist()
    return on7_list


def cons_to_es():
    on7_list = get_on7_list()
    for on7 in on7_list:
        sql = "select * from {c_cons_all} where on7={on7}".format(
            c_cons_all=c_cons_all, on7=on7
        )
        df = sql_engine.query(sql)
        print(len(df))
        ep.to_es(df, index_name, request_timeout=60)


if __name__ == "__main__":
    c_cons_all = config.get("C_CONS_ALL", "aclr_base_doc_all")
    index_name = c_cons_all + "-" + datetime.today().strftime("%Y.%m.%d")

    ep = es_pandas(config["ES_HOST"])
    print(f"mysql全量用户档案表:{c_cons_all}，同步至es索引：{index_name}")
    cons_to_es()
    print(f"索引:{index_name}，修改别名为：{c_cons_all}")
    modify_alias(index_name, c_cons_all)
    print("end")
